package cn.z2huo.demo.kafka.spring.retry.nonblock.config.cluster1;

import cn.z2huo.demo.kafka.spring.retry.nonblock.properties.KafkaConfiguration;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
import org.springframework.kafka.retrytopic.RetryTopicConfigurationBuilder;
import org.springframework.retry.backoff.BackOffPolicyBuilder;
import org.springframework.retry.backoff.SleepingBackOffPolicy;

import java.time.Duration;
import java.util.List;

/**
 * <p>
 *
 * @author z2huo
 */
@Configuration
@RequiredArgsConstructor
@EnableConfigurationProperties(KafkaConfiguration.class)
public class KafkaRetryConfig {

    private final KafkaConfiguration kafkaConfiguration;

    @Bean
    public RetryTopicConfiguration kafkaRetryTopicConfiguration(
            @Qualifier("cluster1KafkaTemplate") KafkaTemplate<String, String> kafkaTemplate,
            @Qualifier("cluster1KafkaListenerContainerFactory")ConcurrentKafkaListenerContainerFactory<String, String> containerFactory
            ) {
        KafkaProperties.Retry.Topic retryTopicProperties = kafkaConfiguration.getCluster1().getRetry().getTopic();
        RetryTopicConfigurationBuilder builder = RetryTopicConfigurationBuilder.newInstance()
                .maxAttempts(retryTopicProperties.getAttempts())
                // 相同间隔使用同一主题
                .useSingleTopicForSameIntervals()
                // 重试主题以索引为后缀
                .suffixTopicsWithIndexValues()
                // 不自动创建重试主题
//                .doNotAutoCreateRetryTopics()
                // 自动创建主题，并指定分区和复制因子
                .autoCreateTopicsWith(5, (short) 3)
                // 指定要重试的主题
                .includeTopics(List.of(
                        kafkaConfiguration.getCluster1Topic().get("retry-test"),
                        kafkaConfiguration.getCluster1Topic().get("temp")
                ))
                // 指定要排除的主题
                .excludeTopics(List.of(
                        kafkaConfiguration.getCluster1Topic().get("no-retry")
                ))
                .listenerFactory(containerFactory)
                // 指定 dlt 处理程序方法
                .dltHandlerMethod("customDltProcessor", "processDltMessage")
                // 配置如果 dlt 处理方法也处理失败时的行为
                .doNotRetryOnDltFailure()
                ;

        this.setBackOffPolicy(builder, retryTopicProperties);

        RetryTopicConfiguration retryTopicConfiguration = builder.create(kafkaTemplate);

        return retryTopicConfiguration;
    }

    private void setBackOffPolicy(RetryTopicConfigurationBuilder builder,
                                  KafkaProperties.Retry.Topic retryTopicProperties) {
        long delay = (retryTopicProperties.getDelay() != null) ? retryTopicProperties.getDelay().toMillis() : 0;
        if (delay > 0) {
            PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
            BackOffPolicyBuilder backOffPolicy = BackOffPolicyBuilder.newBuilder();
            map.from(delay).to(backOffPolicy::delay);
            map.from(retryTopicProperties.getMaxDelay()).as(Duration::toMillis).to(backOffPolicy::maxDelay);
            map.from(retryTopicProperties.getMultiplier()).to(backOffPolicy::multiplier);
            map.from(retryTopicProperties.isRandomBackOff()).to(backOffPolicy::random);
            builder.customBackoff((SleepingBackOffPolicy<?>) backOffPolicy.build());
        }
        else {
            builder.noBackoff();
        }
    }

}
